package cn.gwm.flink.streaming.ods.model;

import cn.gwm.flink.streaming.beans.BeanSource;
import cn.gwm.flink.streaming.constant.BaseFields;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.orc.vector.Vectorizer;
import org.apache.hadoop.hive.ql.exec.vector.BytesColumnVector;
import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;

import java.io.IOException;
import java.io.Serializable;
import java.nio.charset.StandardCharsets;
import java.util.List;

/**
 * @ClassName OdsOrcVectorizer
 * @Author xzh
 * @Date 2022/12/20 16:02
 **/
@Slf4j
public class OdsOrcVectorizer extends Vectorizer<JSONObject> implements Serializable {

    private BeanSource.SourceEnum sourceEnum;

    private List<String> list;

    public OdsOrcVectorizer(String schema, BeanSource.SourceEnum sourceEnum) {
        super(schema);
        this.sourceEnum = sourceEnum;
    }
    public OdsOrcVectorizer(String schema,List<String> list) {
        super(schema);
        this.list = list;
    }

    @Override
    public void vectorize(JSONObject object, VectorizedRowBatch batch) throws IOException {
        try {
            List<String> list;
            if(this.list!=null){
                list = this.list;
            }else{
                list = BeanSource.sourceList(sourceEnum);
            }
            int size = list.size();
            if (list.contains(BaseFields.PARTITION_KEY)) {
                size = size - 1;
            }
            int rowId = batch.size++;
            for (int i = 0; i < size; i++) {
                BytesColumnVector colVector = (BytesColumnVector) batch.cols[i];
                Object v = object.get(list.get(i));
                String value = "";
                if (!object.containsKey(list.get(i)) || StrUtil.DASHED.equals(v)) {
                    value = BaseFields.defaultVal;
                } else {
                    value = v.toString();
                }
                colVector.setVal(rowId, value.getBytes(StandardCharsets.UTF_8));
            }
        }catch (Exception e){
            e.printStackTrace();
            log.error("orc 解析数据填充异常，object={}",object.toString());
            throw e;
        }
    }
}
